package cn.ichiva.service;

import cn.ichiva.common.ThreadContent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;

import javax.websocket.Session;
import java.io.IOException;
import java.util.Date;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 订阅/发布
 * 只支持websocket链接
 * 未完成
 */
@Slf4j
public class MessageService {

    private final RedisMessageListenerContainer container;
    private final RedisTemplate<String,String> redisTemplate;

    //长时间不推送数据自动注销未完成
    private final ConcurrentHashMap<String, SessionMessageListenerAdapter> map;

    public MessageService(RedisConnectionFactory connectionFactory, RedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        this.map = new ConcurrentHashMap<>();
    }

    public String addMessageListener(String account){
        String topic = UUID.randomUUID().toString();
        SessionMessageListenerAdapter listener = new SessionMessageListenerAdapter(account);
        map.put(topic,listener);
        container.addMessageListener(listener,new PatternTopic(topic));
        return topic;
    }

    public void removeMessageListener(String account, String topic){
        SessionMessageListenerAdapter listener = map.get(topic);
        if(listener != null && listener.getAccount().equals(account)){
            container.removeMessageListener(listener,new PatternTopic(topic));
        }
    }

    public void join(String topic,Session session){
        SessionMessageListenerAdapter listener = map.get(topic);
        if(listener != null) listener.join(session);
    }

    public void quit(String topic,Session session){
        SessionMessageListenerAdapter listener = map.get(topic);
        if(listener != null) listener.quit(session);
    }

    public void push(String topic, String message) {
        redisTemplate.convertAndSend(topic,message);
    }

    static class SessionMessageListenerAdapter extends MessageListenerAdapter{
        private final Map<Session,Date> sessions =  new ConcurrentHashMap<>();
        private final String account;

        public SessionMessageListenerAdapter(String account) {
            this.account = account;
        }

        public void join(Session session){
            sessions.put(session,new Date());
        }

        public void quit(Session session){
            sessions.remove(session);
        }

        public String getAccount() {
            return account;
        }

        @Override
        public void onMessage(Message message, byte[] pattern) {
            String text = new String(message.getBody());
            sessions.forEach((session, date) -> {
                try {
                    session.getBasicRemote().sendText(text);
                } catch (IOException e) {
                    log.debug("消息推送失败,sessionId = {},{}",session.getId(),text);
                }
            });
        }
    }
}
